#pragma once
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "mq_threadpool.hpp"
#include "mq_router.hpp"
#include "mq_virhost.hpp"
#include "../mqcommon/mq_proto.pb.h"
#include "mq_consumer.hpp"
#include <condition_variable>
namespace mymq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>; // 通用响应
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    class Channel
    {
    public:
        using Ptr = std::shared_ptr<Channel>;
        Channel(muduo::net::TcpConnectionPtr conn,
                ProtobufCodecPtr codec)
            : _cid(GetUUIDHelper::GetUUID()),
              _conn(conn),
              _codec(codec)
        {
        }
        ~Channel()
        {
            BasicCancel();
        }
        bool OpenChannel()
        {
            openChannelRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        void CloseChannel()
        {
            openChannelRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            WaitResponse(rid);
        }
        std::string Cid() { return _cid; }
        // 声明交换机
        bool DeclareExchange(const std::string &ename,
                             const ExchangeType &type,
                             const bool durable,
                             const bool auto_del,
                             google::protobuf::Map<std::string, std::string> &args)
        {
            // 1.先构造请求
            declareExchangeRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_del);
            req.mutable_args()->swap(args);
            // 2.发送请求
            _codec->send(_conn, req);
            // 3.等待响应，确定是否已经完成创建
            basicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        // 删除交换机
        void DeleteExchange(const std::string &ename)
        {
            // 先构造请求
            deleteExchangeRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }
        // 声明队列
        bool DeclareQueue(const std::string &qname,
                          const bool durable,
                          const bool exclusive,
                          const bool auto_delete,
                          google::protobuf::Map<std::string, std::string> &args)
        {
            // 1.先构造请求
            declareQueueRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_durable(durable);
            req.set_exclusive(exclusive);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 2.发送请求
            _codec->send(_conn, req);
            // 3.等待响应，确定是否已经完成创建
            basicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        // 删除队列
        void DeleteQueue(const std::string &qname)
        {
            deleteQueueRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }
        // 队列绑定
        bool QueueBind(const std::string &ename, const std::string &qname, const std::string &binding_key)
        {
            queueBindRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(binding_key);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        // 解除绑定
        void QueueUnBind(const std::string &ename, const std::string &qname)
        {
            queueUnBindRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }
        // 发布消息
        bool BasicPublish(const std::string &ename, const BasicProperties *bp, const std::string &body)
        {
            basicPublishRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery(bp->delivery());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = WaitResponse(rid);
            return resp->ok();
        }
        // 订阅消息
        bool BasicConsume(const std::string &qname, const std::string &tag, const bool auto_ack, const ConsumerCallBack &cb)
        {
            if (_consumer.get() != nullptr)
            {
                DLOG("当前消费者已经订阅了队列");
                return false;
            }
            basicConsumeRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(tag);
            req.set_queue_name(qname);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn, req);
            basicCommonResponsePtr resp = WaitResponse(rid);
            if (resp->ok() == false)
            {
                DLOG("订阅消息失败");
                return false;
            }
            // 创建消费者
            _consumer = std::make_shared<Consumer>(tag, qname, auto_ack, cb);
            return true;
        }
        // 取消订阅
        void BasicCancel()
        {
            if (_consumer.get() == nullptr)
                return;
            basicCancelRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(_consumer->_tag);
            req.set_queue_name(_consumer->_qname);
            _codec->send(_conn, req);
            _consumer.reset();
            WaitResponse(rid);
        }
        // 确认消息
        void BasicAck(const std::string &msgid)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("未找到对应的消费者");
                return;
            }
            basicAckRequest req;
            std::string rid = GetUUIDHelper::GetUUID();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->_qname);
            req.set_message_id(msgid);
            _codec->send(_conn, req);
            WaitResponse(rid);
        }

    public:
        // 接收到基础的响应后向map中添加消息
        void PutBasicResponse(const basicCommonResponsePtr &resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(), resp));
            _cv.notify_all(); // 唤醒所有线程，让线程去查看是否收到对应的响应
        }
        // 消息的推送，连接收到对应的推送消息后，需要通过信道找到对应的消费者，然后进行回调处理
        void Consume(const basicConsumeResponsePtr &resp)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("未找到对应的消费者");
                return;
            }
            if (resp->consumer_tag() != _consumer->_tag)
            {
                DLOG("收到的消息的消费者标识与当前信道消费者不一致");
                return;
            }
            _consumer->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
            return;
        }

    private:
        // 等待某个请求响应
        basicCommonResponsePtr WaitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [&rid, this]()
                     {
                         return _basic_resp.find(rid) != _basic_resp.end(); // 等待成功
                     });
            // 构建请求
            basicCommonResponsePtr resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return resp;
        }

    private:
        std::string _cid;                   // 信道id
        muduo::net::TcpConnectionPtr _conn; // 连接对象
        ProtobufCodecPtr _codec;            // 协议处理对象
        Consumer::Ptr _consumer;            // 订阅者
        // 将收到的响应存放到该容器，因为muduo库中的操作是异步的，只有收到某个请求的响应后才能进行下一步操作
        std::unordered_map<std::string, basicCommonResponsePtr> _basic_resp;
        std::mutex _mutex;
        std::condition_variable _cv;
    };

    class ChannelManager
    {
    public:
        using Ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        //创建信道
        Channel::Ptr CreateChannel(const muduo::net::TcpConnectionPtr conn,
                           const ProtobufCodecPtr codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto channel = std::make_shared<Channel>(conn,codec);
            _channels.insert(std::make_pair(channel->Cid(),channel));
            return channel;
        }
        //删除信道
        void DeleteChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        //获取一个信道
        Channel::Ptr GetChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                DLOG("没有找到该信道");
                return Channel::Ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::Ptr> _channels;
    };
}
